package streamproc.eventfiringcomponents;

import java.util.HashMap;
import java.util.Vector;
import streamdef.StreamDefinitionManager;
import streamproc.Chain;
import streamproc.ChainComponent;
import event.Event;
import event.EventCallbackManager;

// Last component to check if an event should be passed to the EventCallbackManager.
// Each period check if a buffer contains events -> avoid a circular chain this way.
public class EventFiringComponent extends ChainComponent {
	private static EventFiringComponent sInstance = null;

	public static EventFiringComponent instance() {
		if (sInstance == null)
			sInstance = new EventFiringComponent();

		return sInstance;
	}

	private Vector<Event> eventsToFire = null;
	private Vector<Vector<Event>> vectorsOfEventsToFire = null;
	private Vector<Event> eventsToPutBackInChain = null;
	private boolean active = false;

	private EventFiringComponent() {
		eventsToFire = new Vector<Event>();
		vectorsOfEventsToFire = new Vector<Vector<Event>>();
		eventsToPutBackInChain = new Vector<Event>();
		active = false;
	}

	public void reset() {
		if (eventsToFire != null)
			eventsToFire.clear();
		if (vectorsOfEventsToFire != null)
			vectorsOfEventsToFire.clear();
		if (eventsToPutBackInChain != null)
			eventsToPutBackInChain.clear();

		active = false;
	}

	protected void finalize() {
		reset();
	}

	public void init() {
		reset();
		active = true;
	}

	public boolean isActive() {
		return active;
	}

	public synchronized void processEvents(Vector<Event> events) {
		HashMap<Long, Vector<Event>> temp = new HashMap<Long, Vector<Event>>();
		for (int i = 0; i < events.size(); ++i) {
			Event e = events.get(i);

			// fire the event if a callback function is present
			if (EventCallbackManager.instance().isEventCallbackAvailable(
					e.getStreamID()))
				eventsToFire.add(e);

			if (EventCallbackManager.instance().isEventVectorCallbackAvailable(
					e.getStreamID())) {
				if (temp.get(new Long(e.getStreamID())) == null)
					temp.put(new Long(e.getStreamID()), new Vector<Event>());
				temp.get(new Long(e.getStreamID())).add(e);
			}

			// Push back in the chain if it's used in another query
			// Watch out for a circular chain here !!
			if (StreamDefinitionManager.instance()
					.streamIDIsUsedInOtherQueries(e.getStreamID())) {
				eventsToPutBackInChain.add(e);
			}
		}

		for (Object key : temp.keySet()) {
			Vector<Event> v = temp.get(key);
			vectorsOfEventsToFire.add(v);
		}
	}

	public synchronized void releaseFire() {
		// System.out.print(eventsToFire.size() + " ");
		// System.out.println(eventsToPutBackInChain.size() + " ");
		for (int i = 0; i < eventsToFire.size(); ++i) {
			Event e = eventsToFire.get(i);
			EventCallbackManager.instance().fireEvent(e);
		}
		eventsToFire.clear();

		for (int i = 0; i < vectorsOfEventsToFire.size(); ++i) {
			Vector<Event> e = vectorsOfEventsToFire.get(i);
			EventCallbackManager.instance().fireVectorOfEvents(e);
		}
		vectorsOfEventsToFire.clear();

		for (int i = 0; i < eventsToPutBackInChain.size(); ++i) {
			Event e = eventsToPutBackInChain.get(i);
			Chain.instance().processEvent(e);
		}
		eventsToPutBackInChain.clear();
	}
}
